/**
 * Software License, Version 1.0 Copyright 2003 The Trustees of Indiana
 * University. All rights reserved. Redistribution and use in source and binary
 * forms, with or without modification, are permitted provided that the
 * following conditions are met: 1) All redistributions of source code must
 * retain the above copyright notice, the list of authors in the original source
 * code, this list of conditions and the disclaimer listed in this license; 2)
 * All redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the disclaimer listed in this license in the
 * documentation and/or other materials provided with the distribution; 3) Any
 * documentation included with all redistributions must include the following
 * acknowledgement: "This product includes software developed by the Community
 * Grids Lab. For further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/." Alternatively, this acknowledgement may
 * appear in the software itself, and wherever such third-party acknowledgments
 * normally appear. 4) The name Indiana University or Community Grids Lab or
 * NaradaBrokering, shall not be used to endorse or promote products derived
 * from this software without prior written permission from Indiana University.
 * For written permission, please contact the Advanced Research and Technology
 * Institute ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5)
 * Products derived from this software may not be called NaradaBrokering, nor
 * may Indiana University or Community Grids Lab or NaradaBrokering appear in
 * their name, without prior written permission of ARTI. Indiana University
 * provides no reassurances that the source code provided does not infringe the
 * patent or any other intellectual property rights of any other entity. Indiana
 * University disclaims any liability to any recipient for claims brought by any
 * other entity based on infringement of intellectual property rights or
 * otherwise. LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH
 * NO WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY
 * GIVES NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.transport.http;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;

import org.apache.log4j.Logger;

import cgl.narada.transport.TransportException;
import cgl.narada.transport.proxy.ProxyTunnel;
import cgl.narada.transport.sslHttpBase.LinkMessageKeepAlive;
import cgl.narada.transport.sslHttpBase.Transport;
import cgl.narada.transport.sslHttpBase.TransportDataListener;
import cgl.narada.transport.sslHttpBase.TransportImpl;
import cgl.narada.util.ByteUtilities;

/**
 * @author Hongbin Liu
 * @author Harshawardhan Gadgil
 * @version 1.1
 */

public class HTTPTransportImpl extends TransportImpl {
	final static Logger log = Logger.getLogger(HTTPTransportImpl.class
			.getName());

	private int transactionNum = 0; // random?

	// Beyond this size means the message is corrupt.
	private static final int MAX_PAYLOAD_SIZE = 10000000;
	private Socket socket = null;

	// Indicated whether disconnect caused by io errors.
	private boolean ioerror = false;
	// private SocketFactory httpFactory = null;
	// private static Random s_rand = new Random(System.currentTimeMillis());
	//	
	private HTTPReceiverThread m_receiverThread = new HTTPReceiverThread(
			"httpreceiver");
	private HTTPSenderThread m_senderThread = new HTTPSenderThread("httpsender");

	private HTTPDeliveryThread m_deliveryThread = null;

	private TransportDataListener m_listener = null;

	public static final int MAX_POST_SIZE = 20;
	public static final long PUSH_INTERVAL = 50;

	// transport factory server listening port; should make ONE variable
	// public static final int DEFAULT_HTTP_PORT = 12349;

	/**
	 * -------------------------------------------------------------------
	 */
	public HTTPTransportImpl() {

	}

	/**
	 * ------------------------------------------------------------------- Used
	 * when an HTTP transport end point is accepted rather than initiated.
	 */
	protected HTTPTransportImpl(Socket socket) {
		this.socket = socket;
		try {
			startThreads();
		}
		catch (IOException e) {
			log.warn("HTTPTrasportImpl Unable to start send/receive thread");
		}
	}

	class HTTPSenderThread extends Thread {
		long lastCheck = System.currentTimeMillis();
		long lastStatusCheck = System.currentTimeMillis();
		boolean stopped = false;

		public HTTPSenderThread(String name) {
			super(name);
		}

		public void run() {
			while (!stopped) {
				int numMessages = m_sendQueue.getNumMessages();
				long currentTime = System.currentTimeMillis();
				if (m_statusInterval != 0
						&& currentTime - lastStatusCheck > m_statusInterval) {
					statusCheck();
					lastStatusCheck = currentTime;
				}
				if (numMessages >= MAX_POST_SIZE
						|| currentTime - lastCheck >= PUSH_INTERVAL) {
					lastCheck = currentTime;
					try {
						byte[] messages = m_sendQueue.removeAll();
						if (messages.length == 0) continue;
						// log.debug("send thread messages length " +
						// messages.length);
						sendDataImmediately(messages);
					}
					catch (Exception e) {
						e.printStackTrace();
						handleSendException(e);
					}
				}
				try {
					Thread.sleep(10);
				}
				catch (InterruptedException ie) {
					ie.printStackTrace();
				}
			}
		}

		public void statusCheck() {

			LinkMessageKeepAlive kam = new LinkMessageKeepAlive();
			byte[] bytes = kam.toBytes(); // type, channelId len, channelId,
			// payload len payload
			OutputStream out = null;
			try {
				out = socket.getOutputStream();
			}
			catch (IOException ioe) {
				ioe.printStackTrace();
			}
			int bodyLen = bytes.length;
			String headers = "POST http://" + m_host + ":" + m_port + "/"
					+ (transactionNum++) + "HTTP/1.0\n"
					+ "Proxy-Connection: Keep-Alive\n" + "Progma: no-cache\n"
					+ "Content-Type: application/octet-stream\n"
					+ "Content-Length: " + (bodyLen + 8) + "\r\n\n";
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			/*
			 * When a package of messages is received and put into the queue,
			 * it's assumed be of the following format: [total length][msg1
			 * length][msg1][msg2 length] ..."
			 */
			try {
				if (out != null) {
					baos.write(headers.getBytes());
					baos.write(ByteUtilities.getBytes(bodyLen + 4));
					baos.write(ByteUtilities.getBytes(bodyLen));
					baos.write(bytes);
					out.write(baos.toByteArray());
					log.debug("status check, http message body "
							+ (bodyLen + 8) + " bytes");
				}
			}
			catch (IOException ioe) {
				ioe.printStackTrace();
				log.warn(ioe.getMessage());
			}
		}

		public void destroy() {
			if (!stopped) {
				stopped = true;
				interrupt();
			}
		}

	}

	class HTTPReceiverThread extends Thread {
		boolean stopped = false;

		public HTTPReceiverThread(String name) {
			super(name);
		}

		public void run() {
			while (!stopped) {
				try {
					byte[] body = receiveDataImmediately();
					if (body.length != 0) m_receiveQueue.addAll(body);
				}
				catch (IOException ioe) {
					ioe.printStackTrace();
					handleReceiveException(ioe);
				}
			}
		}

		public void destroy() {
			if (!stopped) {
				stopped = true;
				interrupt();
			}
		}
	}

	/**
	 * =======================================================================
	 * Helper thread to take data off of the receive queue and invoke data
	 * received on the listener.
	 */
	class HTTPDeliveryThread extends Thread {
		private boolean m_stopped = false;
		private Transport m_parent = null;

		HTTPDeliveryThread(Transport parent) {
			m_parent = parent;
			setName("httpdelivery");
		}

		/**
		 *
		 */
		public void run() {
			log.debug("delivery thread starting ... ");
			while (!m_stopped) {
				byte[] data = m_receiveQueue.removeData();
				if (data != null) {
					m_listener.dataReceived(m_parent, data);
				}
				else {
					if (!m_stopped) {
						log
								.warn("TransportImpl Delivery thread received null data.  Exiting");
					}
					destroy();
				}
			}
		}

		/**
		 * 
		 */
		public void destroy() {
			if (!m_stopped) {
				m_stopped = true;
				interrupt();
				// June 30, 2004.
				// Should dispose even there is no delivery thread like in
				// tests.
				// m_receiveQueue.dispose();
			}
		}
	}

	/**
	 * 
	 */
	public void startThreads() throws IOException {
		m_receiverThread.start();
		m_senderThread.start();
	}

	/**
	 * 
	 */
	public void stopThreads() throws IOException {
		m_receiverThread.destroy();
		m_senderThread.destroy();
		if (m_deliveryThread != null) m_deliveryThread.destroy();
		if (m_listener != null) m_listener.transportClosed(this);
	}

	/**
	 * 
	 */
	public void setTransportDataListener(TransportDataListener l) {
		if (m_listener == null && l != null) {
			m_deliveryThread = new HTTPDeliveryThread(this);
			m_deliveryThread.start();
		}
		else if (l == null && m_deliveryThread != null) {
			m_deliveryThread.destroy();
			m_deliveryThread = null;
		}
		m_listener = l;
	}

	/**
	 * 
	 */
	public boolean isConnected() {
		return socket != null;
	}

	/**
	 * 
	 */
	public boolean isInErrorState() {
		return ioerror;
	}

	/**
	 * 
	 */
	public String getType() {
		return HTTPLinkFactory.CHANNEL_TYPE;
	}

	/**
	 * 
	 */
	public void connect() throws TransportException {
		if (m_host == null) {
			throw new TransportException("Unspecified endpoint m_host");
		}
		connect(m_host, m_port);
	}

	/**
	 * 
	 */
	public void connect(String host, int port) throws TransportException {
		String proxyHost = proxyInfo.getProxyHost();
		try {
			if (proxyHost != null) {
				proxyTunnel = new ProxyTunnel(proxyInfo);
				socket = proxyTunnel.doProxyConnect(host, port);
				log.debug("Connected to proxy @" + proxyHost + ":"
						+ proxyInfo.getProxyPort());
			}
			else {
				socket = new Socket(host, port);
				log.debug("Connected to server @: " + host + ":" + port);
			}

			socket.setSoLinger(false, 0);
			socket.setTcpNoDelay(true);

			startThreads();

		}
		catch (IOException e) {
			e.printStackTrace();
			throw new TransportException("connect error: " + e.getMessage());
		}
	}

	/**
	 * 
	 */
	public void connect(String host) throws TransportException {
		connect(host, m_port);
	}

	/**
	 * 
	 */
	public synchronized void disconnect() throws IOException {
		if (socket != null) {
			stopThreads();
			socket.close();
			socket = null;
		}
	}

	/**
	 *
	 */
	public boolean isSecure() {
		return false;
	}

	/**
	 *
	 */
	public String toString() {
		if (m_host != null) {
			return "=>" + m_host + ":" + m_port;
		}
		else {
			return "<=" + socket.getInetAddress().getHostName() + ":"
					+ socket.getLocalPort();
		}
	}

	/**
	 *
	 */
	protected void throwBadConnectionException() throws IOException {
		if (ioerror) {
			throw new IOException("Connection in error state.  disconnected");
		}
		else {
			throw new IOException("Connection closed");
		}
	}

	/**
	 *
	 */
	protected void sendDataImmediately(byte[] messages) throws IOException {

		if (socket == null) throwBadConnectionException();
		OutputStream out = socket.getOutputStream();
		ByteArrayOutputStream baos = new ByteArrayOutputStream();
		transactionNum++;
		int bodyLen = messages.length;
		String headers = "POST http://" + m_host + ":" + m_port + "/"
				+ transactionNum + "HTTP/1.0\n"
				+ "Proxy-Connection: Keep-Alive\n" + "Progma: no-cache\n"
				+ "Content-Type: application/octet-stream\n"
				+ "Content-Length: " + (bodyLen + 4) + "\r\n\n";
		baos.write(headers.getBytes());
		baos.write(ByteUtilities.getBytes(bodyLen));
		baos.write(messages);
		// log.debug("send immd");
		out.write(baos.toByteArray());
		out.flush();
		// log.debug("send immd. 2");
		int i = ByteUtilities.getInt(ByteUtilities.getBytes(bodyLen));
		// log.debug("send imme. sending "+ i);
	}

	/**
	 * @return byte array that possibly include a series of messages with each
	 *         led by 4 bytes of length. We usually don't need to use this since
	 *         everything will be put in the receive queue.
	 */
	protected byte[] receiveDataImmediately() throws IOException {

		if (socket == null) throwBadConnectionException();
		InputStream input = socket.getInputStream();

		int chars = 0;
		while (chars < 3) {
			int ch = input.read();
			if (ch < 0) {
				throw new IOException("Unexpected EOF from peer");
			}
			if (ch == '\r' || ch == '\n') {
				chars++;
			}
			else if (ch != '\r' && ch != '\n') {
				chars = 0;
			}
		}
		byte[] lenBytes = new byte[4];
		input.read(lenBytes);
		int len = ByteUtilities.getInt(lenBytes, 0);
		// log.debug("receive imme. " + len);
		if (len > MAX_PAYLOAD_SIZE) {
			throw new IOException("Bad size or payload too large: " + len);
		}
		byte[] body = readBytes(input, len);
		if (handleTransportSystemMessage(lenBytes)) {
			return new byte[0];
		}
		// log.debug("receive imme. received "+ body.length);
		int cnt = 0;
		for (int pos = 0; pos < body.length; pos += 4) {
			int slen = ByteUtilities.getInt(body, pos);
			// log.debug("receive single message len " + slen);
			pos += slen;
			cnt++;
		}
		// log.debug("receive imme count = " + cnt);
		return body;
	}

	/**
	 *
	 */
	protected void handleReceiveException(Exception e) {
		// Handle by closing the connection.
		try {
			ioerror = true;
			// June 30, 2004. Move it from delivery thread here.
			m_receiveQueue.dispose();

			disconnect();
		}
		catch (IOException ex) {
			ex.printStackTrace();
		}
	}

	/**
	 *
	 */
	protected void handleSendException(Exception e) {
		// Handle by closing the connection.
		try {
			ioerror = true;
			disconnect();
		}
		catch (Exception ex) {
			ex.printStackTrace();
		}
	}

	/**
	 *
	 */
	private byte[] readBytes(InputStream input, int size) throws IOException {
		byte[] buf = new byte[size];

		int remainingSize = size;
		int readSize = 0;
		while ((readSize = input.read(buf, (size - remainingSize),
				remainingSize)) < remainingSize) {
			if (readSize < 0) {
				throw new IOException(
						"Stream terminated before chunk can be fully read[size="
								+ size + "]");
			}
			remainingSize -= readSize;
		}
		return buf;
	}
}
